package com.runa.monitor.platinterface.netservice.connect;

import com.runa.monitor.platinterface.base.BaseContainer;
import com.runa.monitor.platinterface.netservice.connect.commstatus.CommStatus;
import com.runa.monitor.platinterface.netservice.equipment.PlcService;
import com.runa.protocol.dto.IRequestData;
import com.runa.protocol.dto.IResponseData;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.AbstractMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 真·异步缓存-优先级-批次处理会话
 *
 */
public class AsyncCacheSession implements Session {

    private static final Logger logger = LoggerFactory.getLogger(AsyncCacheSession.class);

    private Channel channel;

    private PlcService service;

    private int timeout;

    private Map<Short, Map.Entry<IRequest<IRequestData>, Long>> cache = new ConcurrentHashMap<>();

    private Map<Integer, Queue<IRequest<IRequestData>>> orderRequestQueue = new ConcurrentSkipListMap<>();

    public AsyncCacheSession(PlcService service, int timeout) {
        this.service = service;
        this.timeout = timeout == 0 ? 1000 :timeout;
    }

    public void pushRequest(IRequest<IRequestData> request) {
        Queue<IRequest<IRequestData>> requestQueue = orderRequestQueue.get(request.flag(0));
        short sign = incrementShort();
        request.getRequestData().setTransSign(sign);
        if (requestQueue == null) {
            requestQueue = new ConcurrentLinkedQueue<>();
            orderRequestQueue.put(request.flag(0), requestQueue);
        }
        requestQueue.offer(request);
    }

    @Override
    public void linked(Channel channel) {
        if (channel != null && channel.isActive()) {
            cache.clear();
//            closeAndRemove();
            this.channel = channel;
            BaseContainer.putSession(channel.id().asLongText(), this);
            setOnlineStatus(true);
        }
    }

    @Override
    public void requesting() throws InterruptedException {
        if (isOnline()) {
            IRequest<IRequestData> currRequest = null;
            for (Queue<IRequest<IRequestData>> queue : orderRequestQueue.values()) {
                if (!queue.isEmpty()) {
                    currRequest = queue.poll();
                    break;
                }
            }
            if (currRequest != null) {
                TimeUnit.MILLISECONDS.sleep(cache.entrySet().size() * 50);
                cache.put(currRequest.getRequestData().getTransSign(),
                        new AbstractMap.SimpleEntry<>(currRequest, System.currentTimeMillis()));
                channel.writeAndFlush(currRequest.getRequestData());
            }
            Iterator<Map.Entry<Short, Map.Entry<IRequest<IRequestData>, Long>>> iterator = cache.entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<Short, Map.Entry<IRequest<IRequestData>, Long>> next = iterator.next();
                if (System.currentTimeMillis() - next.getValue().getValue() > timeout) {
                    iterator.remove();
                    reset(CommStatus.TimeOut, next.getValue().getKey());
                }
            }
        }
    }

    private void reset(CommStatus status, IRequest<IRequestData> currRequest) {
        logger.info("Session Reset PLC:{} {} --- {}", service.getId(), status);
        Queue<IRequest<IRequestData>> requestQueue = orderRequestQueue.get(currRequest.flag(0));
        requestQueue.removeIf(req -> req.flag(1) == currRequest.flag(1));
        currRequest.cancel(status);
    }

    @Override
    public void receiving(IResponseData data) {
        if (data != null) {
            Map.Entry<IRequest<IRequestData>, Long> entry = cache.remove(data.getTransSign());
            if (entry != null) {
                IRequest<IRequestData> request = entry.getKey();
                request.getResponse().setResponseData(data);
                if (request.check()) {
                    request.getResponse().doReceive();
                }
            }
        }
    }

    @Override
    public void setOnlineStatus(boolean status) {
        this.service.updateCommStatus(status);
    }

    @Override
    public boolean isOnline() {
        return channel != null && channel.isActive();
    }

    @Override
    public boolean close() {
        try {
            if (channel != null && channel.isActive()) {
                channel.close().sync();
            }
            channel = null;
        } catch (Exception e) {
            return false;
        }
        return true;
    }

    private AtomicInteger incrSign = new AtomicInteger();

    private short incrementShort() {
        return (short) incrSign.getAndIncrement();
    }

    @Override
    public String toString() {
        return service.getId()+"'s session";
    }
}
